package com.xinqi.modules.exam.common.delayed.consumer;

import com.xinqi.common.base.utils.json.JSON;
import com.xinqi.common.rocketmq.annotation.MqConsumer;
import com.xinqi.common.rocketmq.util.ByteUtils;
import com.xinqi.modules.exam.common.delayed.api.MqQueueHandler;
import com.xinqi.modules.exam.common.delayed.dto.ExamMqMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 * @author linkai
 * @date 2023/11/23
 */
@Slf4j
@Component
@MqConsumer(topic = "${rocketmq.examTopic}", groupName = "exam_delay")
@RequiredArgsConstructor
public class ExamMqMessageConsumer implements MessageListener {
    private final List<MqQueueHandler> handlers;
    @Value("${rocketmq.examTopic}")
    public String examTopic;

    @Override
    public ConsumeResult consume(MessageView messageView) {
        String body = ByteUtils.byteBufferToStr(messageView.getBody());
        log.info("====> 接收到测评消息, topic: {}, id: {}, group: {}, body: {}",
            messageView.getTopic(), messageView.getMessageId(), messageView.getMessageGroup(), body);
        ExamMqMessage message = JSON.parse(body, ExamMqMessage.class);
        if (message != null) {
            if (handler(message)) {
                return ConsumeResult.SUCCESS;
            } else {
                return ConsumeResult.FAILURE;
            }
        } else {
            log.info("====> 测评任务处理失败，消息体为空, topic: {}, id: {}, key: {}, group: {}",
                messageView.getTopic(), messageView.getMessageId(), messageView.getKeys(), messageView.getMessageGroup());
        }
        return ConsumeResult.SUCCESS;
    }

    public boolean handler(ExamMqMessage message) {
        log.info("====> 测评任务处理开始, task: {}, queueName: {}", message.getExamId(), message.getExamQueue());
        try {
            handlers.stream().filter(v -> Objects.equals(v.queueName(), message.getExamQueue())).findFirst().ifPresent(v -> {
                v.handleTask(message.getExamId());
            });
            return true;
        } catch (Exception e) {
            log.error("====> 测评任务处理失败, task: {}, queueName: {}", message.getExamId(), message.getExamQueue(), e);
            return false;
        }
    }
}
